package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

object Demo16Cache {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("map")


    val sc = new SparkContext(conf)

    val student: RDD[String] = sc.textFile("spark/data/students.txt")

    val mapStudent: RDD[String] = student.map(line => {
      println("map----")
      line
    })


    /**
      * 对多次使用的RDD进行缓存， 默认cache是将数据缓存在内存中
      *
      */
    //  mapStudent.cache()

    /**
      * 持久化级别选择
      * 1、如果rdd的数据量不大 ，没有超过内存的限制--->  MEMORY_ONLY
      * 2、如果数据操超过了内存的限制 ---> MEMORY_AND_DISK_SER (压缩之后放内存内存放不下放磁盘)
      * 不管压不压缩放内存都比放磁盘要块，所有尽量将数据压到内存中
      *
      *
      * 缓存实际上是将数据缓存在执行task所在的服务器的内存或者磁盘上
      *
      */

    mapStudent.persist(StorageLevel.MEMORY_AND_DISK_SER)


    //统计班级的人数
    val clazzKVRDD: RDD[(String, Int)] = mapStudent.map(line => (line.split(",")(4), 1))
    val clazzNumRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey(_ + _)

    clazzNumRDD.foreach(println)


    //统计性别的人数


    val genderKVRDD: RDD[(String, Int)] = mapStudent.map(line => (line.split(",")(3), 1))
    val genderNumRDD: RDD[(String, Int)] = genderKVRDD.reduceByKey(_ + _)

    genderNumRDD.foreach(println)


  }

}
